package drds.binlog.client.impl;

import drds.binlog.client.NodeAccessStrategy;
import drds.binlog.common.utils.JsonUtils;
import drds.binlog.common.zookeeper.ZooKeeperClient;
import drds.binlog.common.zookeeper.Zookeepers;
import drds.binlog.common.zookeeper.running.ServerRunningData;
import drds.binlog.protocol.exception.ClientException;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.apache.commons.lang.StringUtils;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * 集群模式的调度策略
 */
public class ClusterNodeAccessStrategy implements NodeAccessStrategy
{

    private IZkChildListener childListener;                                      // 监听所有的服务器列表
    private IZkDataListener dataListener;                                       // 监听当前的工作节点
    private ZooKeeperClient zooKeeperClient;
    private volatile List<InetSocketAddress> inetSocketAddressList = new ArrayList<InetSocketAddress>();
    private volatile InetSocketAddress runningAddress = null;

    public ClusterNodeAccessStrategy(String destination, ZooKeeperClient zooKeeperClient)
    {
        this.zooKeeperClient = zooKeeperClient;
        childListener = new IZkChildListener()
        {

            public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception
            {
                initClusters(currentChilds);
            }

        };

        dataListener = new IZkDataListener()
        {

            public void handleDataDeleted(String dataPath) throws Exception
            {
                runningAddress = null;
            }

            public void handleDataChange(String dataPath, Object data) throws Exception
            {
                initRunning(data);
            }

        };

        String clusterPath = Zookeepers.getDestinationClusterRoot(destination);
        this.zooKeeperClient.subscribeChildChanges(clusterPath, childListener);
        initClusters(this.zooKeeperClient.getChildren(clusterPath));

        String runningPath = Zookeepers.getDestinationServerRunning(destination);
        this.zooKeeperClient.subscribeDataChanges(runningPath, dataListener);
        initRunning(this.zooKeeperClient.readData(runningPath, true));
    }

    @Override
    public SocketAddress currentNode()
    {
        return nextNode();
    }

    public SocketAddress nextNode()
    {
        if (runningAddress != null)
        {// 如果服务已经启动，直接选择当前正在工作的节点
            return runningAddress;
        } else if (!inetSocketAddressList.isEmpty())
        { // 如果不存在已经启动的服务，可能服务是一种lazy启动，随机选择一台触发服务器进行启动
            return inetSocketAddressList.get(0);// 默认返回第一个节点，之前已经做过shuffle
        } else
        {
            throw new ClientException("no alive canal server");
        }
    }

    private void initClusters(List<String> currentChilds)
    {
        if (currentChilds == null || currentChilds.isEmpty())
        {
            inetSocketAddressList = new ArrayList<InetSocketAddress>();
        } else
        {
            List<InetSocketAddress> addresses = new ArrayList<InetSocketAddress>();
            for (String address : currentChilds)
            {
                String[] strs = StringUtils.split(address, ":");
                if (strs != null && strs.length == 2)
                {
                    addresses.add(new InetSocketAddress(strs[0], Integer.valueOf(strs[1])));
                }
            }

            Collections.shuffle(addresses);
            inetSocketAddressList = addresses;// 直接切换引用
        }
    }

    private void initRunning(Object data)
    {
        if (data == null)
        {
            return;
        }

        ServerRunningData runningData = JsonUtils.unmarshalFromByte((byte[]) data, ServerRunningData.class);
        String[] strs = StringUtils.split(runningData.getAddress(), ':');
        if (strs.length == 2)
        {
            runningAddress = new InetSocketAddress(strs[0], Integer.valueOf(strs[1]));
        }
    }

    public ZooKeeperClient getZooKeeperClient()
    {
        return zooKeeperClient;
    }

    public void setZooKeeperClient(ZooKeeperClient zooKeeperClient)
    {
        this.zooKeeperClient = zooKeeperClient;
    }

}
